package com.feiwei

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.table.api.{EnvironmentSettings, Tumble}
import org.apache.flink.table.api.scala.StreamTableEnvironment
import org.apache.flink.types.Row
import org.apache.flink.table.api.scala._
import org.apache.flink.streaming.api.scala._

object day11_GroupWindow {

  def main(args: Array[String]): Unit = {


    val environment = StreamExecutionEnvironment.getExecutionEnvironment
    environment.setParallelism(1)
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)


    val settings = EnvironmentSettings.newInstance()
      .useBlinkPlanner()
      .inStreamingMode().build()
    val path = "E:\\repository\\company\\myself\\flink-learning\\flink-learning-demo\\src\\main\\resources\\sensor.txt"

    val tableEnv = StreamTableEnvironment.create(environment, settings)

    val readStream = environment.readTextFile(path)

    val sensor = readStream.map(v => {

      val arr = v.split(",")

      SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
    })
      //设置时间和水位线,设置为乱序，延迟时间为1秒
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(1)) {
      override def extractTimestamp(t: SensorReading): Long = t.timestamp * 1000
    })


    //如果流中指定了水位线，在转换成表的时候，直接就可以使用字段.rowtime
    var sensortable = tableEnv.fromDataStream(sensor, 'id, 'temperature, 'timestamp.rowtime as 'tp)


    //1使用api方式分组窗口
    var apires = sensortable.window(Tumble over 10.minutes on 'tp as 'tw)
      .groupBy('id, 'tw)
      .select('id, 'id.count, 'tw.end)


    //2 使用sql
    tableEnv.createTemporaryView("sensor", sensortable)

    var sqlres = tableEnv.sqlQuery(
      """
        |select id,
        |count(id) as ct,
        |avg(temperature) as v1,
        |
        |tumble_end(tp,interval '10' second)
        |from sensor
        |group by
        |id,
        |
        |tumble(tp,interval '10' second)
        |
        |
        |
      """.stripMargin)


    sensortable.toAppendStream[Row].print()

    apires.toAppendStream[Row].print("api")
    sqlres.toAppendStream[Row].print("sql")

    environment.execute()

  }
}
